package com.dmp.storm.bolt.db;

import com.dmp.util.DateUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.*;

/** 
 * @ClassName InsertBatchDBBolt
 * @Description 批量插入普通方式
 *
 * @author  sunzy 
 * @date 2016年4月13日 下午4:54:21 
 * @version 1.0 
 *
 */
@SuppressWarnings("serial")
public class InsertBatchDynTableBolt extends MapCacheBolt {
	private static final Logger logger = LoggerFactory.getLogger(InsertBatchDynTableBolt.class);

	private Connection connection;
	//数据库配置
	private String driverClaz = "com.mysql.jdbc.Driver";
    private String url;
    private String user;
    private String pwd;
    
	private String sql;
	private String tableNamePrefix = "";
	private String dateFieldName = "";

	//数据过滤配置
	private String filterFieldName = "";
	private String filterFieldValue = "";
	private List<String> filterList = new ArrayList<String>();
		
    protected String keys;
    protected String outputKeys;
    protected String extraKeys;

    @Override
    public void prepare() {
    	super.prepare();
    	try {
        	Class.forName(getDriverClaz());
            connection = DriverManager.getConnection(url, user, pwd);
            
            //
            if(StringUtils.isNotBlank(getFilterFieldName()) && StringUtils.isNotBlank(getFilterFieldValue())){
            	String[] filters = StringUtils.split(getFilterFieldValue(), ",");
            	filterList = Arrays.asList(filters);
            }
        } catch (Exception e) {
        	logger.error(e.getMessage());
        	
            System.exit(-1);
        }
    }
    
    @Override
    public void processBatch(List<Map<String, Object>> list) {
        Statement stmt = null;
        try {
            stmt = connection.createStatement();
            for (Map<String, Object> map : list) {
                String sql = createSQLByMap(keys, outputKeys, extraKeys, map);
                if(StringUtils.isNotBlank(sql)){
                	stmt.addBatch(sql);
                }
            }
            logger.error(Arrays.toString(stmt.executeBatch()));
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (stmt != null) {
                try {
                    stmt.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
        // 暂时无返回值
    }

    /**
     * @param tableName
     * @param keys
     * @param outputKeys
     * @param map
     * @return
     * @Title: createSQLByMap
     * @Description: 拼接成sql语句
     * @author suzg
     * @date 2015年10月23日 下午3:57:55
     */
    public String createSQLByMap(String keys, String outputKeys, String extraKeys,
                                 Map<String, Object> map) throws Exception {
    	String tableName = buildTableName(map);
    	
    	StringBuilder sb = new StringBuilder();
    	if(StringUtils.isNotBlank(tableName)){
    		String[] keysArray = outputKeys.split(",");
            sb.append("INSERT INTO ").append(tableName).append(" (").append(keys).append(") VALUES(");
            // 组装map中的参数
            for (String key : keysArray) {
                String tempArray[] = key.split("=");// 判断类型
                String realKey = tempArray[0].trim();
                Object obj = map.get(realKey);
                // 处理每一个outputKeys，
                sb = everyKeyJoinToSB(sb, tempArray, obj);
            }
            // 去掉末尾的逗号“，”
            sb = new StringBuilder(sb.toString().substring(0, sb.toString().length() - 1));
            sb.append(")");
    	}
        
        return sb.toString();
    }
    
    private String buildTableName(Map<String, Object> map){
		String tableName = "";
		try{
			if(StringUtils.isBlank(getFilterFieldName()) || 
					(StringUtils.isNotBlank(getFilterFieldName()) 
							&& StringUtils.isNotBlank(getFilterFieldValue()) && filterList.contains("" + map.get(getFilterFieldName())))){
				if(StringUtils.isNotBlank(getDateFieldName())){
					Object dateValue = map.get(getDateFieldName());
					//约定日期字段为字符类型
					if(dateValue != null){
						String dateStr = StringUtils.substring(dateValue.toString(), 0, StringUtils.indexOf(dateValue.toString(), "."));
						Date recordDate = DateUtils.parseDate(dateStr, "yyyy-MM-dd HH:mm:ss");
						tableName = getTableNamePrefix() + DateUtils.formatDate(recordDate, "yyyyMMdd");
					}
				}
			}
		} catch(Exception e){
			logger.error(e.getMessage());
		}
		
		return tableName;
	}

    /**
     * @param sb
     * @param tempArray
     * @param obj
     * @return
     * @Title: everyKeyJoinToSB
     * @Description: 处理每一个outputKeys，以后不同的类型处理时，只要扩展case，处理就可以
     * @author suzg
     * @date 2015年10月26日 下午2:12:56
     */
    private StringBuilder everyKeyJoinToSB(StringBuilder sb, String[] tempArray, Object obj) {
        if (tempArray.length == 1) {
            // 默认是字符串
            sb.append("\'");
            sb.append(obj == null ? null : obj.toString());
            sb.append("\'").append(",");
        } else {
            String type = tempArray[1].trim();// 获取数据类型
            switch (type) {
                case "0":
                    sb.append("\'");
                    // 字符串
                    sb.append(obj == null ? null : obj.toString());
                    sb.append("\'").append(",");
                    break;
                case "1":
                    sb.append("\'");
                    // number
                    sb.append(obj == null ? null : obj.toString());
                    sb.append("\'").append(",");
                    break;

                default:
                    sb.append("\'");
                    sb.append(obj == null ? null : obj.toString());
                    sb.append("\'").append(",");
            }
        }

        return sb;
    }

    public void setKeys(String keys) {
        this.keys = keys;
    }

    public void setOutputKeys(String outputKeys) {
        this.outputKeys = outputKeys;
    }

    public void setExtraKeys(String extraKeys) {
        this.extraKeys = extraKeys;
    }

	public String getDriverClaz() {
		return driverClaz;
	}

	public void setDriverClaz(String driverClaz) {
		this.driverClaz = driverClaz;
	}

	public String getUrl() {
		return url;
	}

	public void setUrl(String url) {
		this.url = url;
	}

	public String getUser() {
		return user;
	}

	public void setUser(String user) {
		this.user = user;
	}

	public String getPwd() {
		return pwd;
	}

	public void setPwd(String pwd) {
		this.pwd = pwd;
	}

	public String getSql() {
		return sql;
	}

	public void setSql(String sql) {
		this.sql = sql;
	}

	public String getTableNamePrefix() {
		return tableNamePrefix;
	}

	public void setTableNamePrefix(String tableNamePrefix) {
		this.tableNamePrefix = tableNamePrefix;
	}

	public String getDateFieldName() {
		return dateFieldName;
	}

	public void setDateFieldName(String dateFieldName) {
		this.dateFieldName = dateFieldName;
	}

	public String getKeys() {
		return keys;
	}

	public String getOutputKeys() {
		return outputKeys;
	}

	public String getExtraKeys() {
		return extraKeys;
	}

	public String getFilterFieldName() {
		return filterFieldName;
	}

	public void setFilterFieldName(String filterFieldName) {
		this.filterFieldName = filterFieldName;
	}

	public String getFilterFieldValue() {
		return filterFieldValue;
	}

	public void setFilterFieldValue(String filterFieldValue) {
		this.filterFieldValue = filterFieldValue;
	}
    
}
